/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.handler;

import java.util.ArrayList;
import java.util.List;
import org.apache.log4j.Logger;
import cn.ac.iie.di.ban.commons.RuntimeEnv;
import cn.ac.iie.di.ban.metastore.Load2DBEntity;
import cn.ac.iie.di.ban.tools.DataBuffer;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 *
 * @author sxl
 */
public abstract class SE2DBWorker implements Runnable {

    private static org.apache.log4j.Logger logger = Logger.getLogger(SE2DBWorker.class.getName());
    protected HttpSe2DBHandler httpSe2DBHandler;
    protected Load2DBEntity load2DBEntity;
    int batchSize;
    List<Map<String, Object>> docRecordSet = new ArrayList();
    long lastUpdateTime = -1;
    int timeout = -1;
    protected volatile boolean stopped;
    protected String dwType;
    protected Long start_time = -1L;
    protected Long end_time = -1L;
    private List<String> result = new ArrayList<>(1);

    public SE2DBWorker(HttpSe2DBHandler httpSe2DBHandler, String pDWType) {
        this.httpSe2DBHandler = httpSe2DBHandler;
        batchSize = httpSe2DBHandler.getLoadNum();
        load2DBEntity = httpSe2DBHandler.getLoad2MBEntity();
        dwType = pDWType;
        stopped = false;
        timeout = (Integer) RuntimeEnv.getParam(RuntimeEnv.TIMEOUT);
    }

    public void run() {
        Thread.currentThread().setName("load msg " + dwType + " worker_" + System.nanoTime());
        long currentTime = System.currentTimeMillis();
        lastUpdateTime = System.currentTimeMillis();
        while (true) {
            List<Map<String, Object>> data = httpSe2DBHandler.getData();
            if (data == null) {
                if (stopped) {
                    freeSource();
                    logger.info("prepare to exit...");
                    break;
                }
                try {
                    Thread.sleep((Integer) RuntimeEnv.getParam(RuntimeEnv.WAIT_TIME));
                } catch (Exception ex) {
                }
            } else {
//                if (lastUpdateTime == -1) {
//                    lastUpdateTime = System.currentTimeMillis();
//                }
                loadData(data);
            }

//            currentTime = System.currentTimeMillis();
//            if (currentTime - lastUpdateTime >= timeout) {
//                logger.info("time out and load data forcelly now:" + new Date(currentTime) + " ago:" + new Date(lastUpdateTime));
//                loadData0();
//                lastUpdateTime = System.currentTimeMillis();
//                logger.info("time out process successfully " + new Date(lastUpdateTime));
//            }
        }
        logger.info(Thread.currentThread().getName() + " exited");
    }

    public void shutdown() {
        stopped = true;
    }

    public abstract void freeSource();

    public void loadData(List<Map<String, Object>> pRecordSet) {
        docRecordSet.addAll(pRecordSet);
//        if (docRecordSet.size() >= load2DBEntity.getLoadNum()) {
            loadData0();
            docRecordSet.clear();
//        }
    }

    public abstract void init() throws Exception;

    public abstract void loadData0();

        public static List<Map<String, String>> yclMsg(Map<String, Object> msg) {
        List<Map<String, String>> returnList = new ArrayList();
        String opinion = msg.get("opinion").toString();
        String channel = msg.get("channel").toString();
        String table = msg.get("table").toString();
        String admin="", agency="", department="", uid = "";
        try {
            String[] req_ids = DataBuffer.findReqId(msg.get("task_id").toString());
            for (String req_id : req_ids) {
                Map<String, String> returnMap = new HashMap<>();
                uid = DataBuffer.getUid(req_id);
                Map<String, Object> agencyMap = DataBuffer.getId4Channel(uid, opinion, channel, table);
                admin = agencyMap.get("admin").toString();
                agency = agencyMap.get("agency").toString();
                department = agencyMap.get("department").toString();
                String opinion_field = agencyMap.get("opinion_field").toString();
                String source_type = agencyMap.get("source_type").toString();
                String source_tunnel = agencyMap.get("source_tunnel").toString();
                returnMap.put("agency", agency);
                returnMap.put("admin", admin);
                returnMap.put("department", department);
                returnMap.put("uid", uid);
                returnMap.put("task_id", msg.get("task_id").toString());
                returnMap.put("scope", msg.get("scope").toString());
                returnMap.put("opinion_field", opinion_field);
                returnMap.put("source_tunnel", source_tunnel);
                returnMap.put("source_type", source_type);
                returnMap.put("data_size", msg.get("data_size").toString());
                returnMap.put("rec_num", msg.get("rec_num").toString());
                returnList.add(returnMap);
            }
            return returnList;
        } catch (Exception e) {
            logger.info(e.getMessage(), e);
        }
        return returnList;
    }
    
    public static void shuntmsg(List<Map<String, String>> docRecords, List<Map<String, String>> recNumMsgs, List<Map<String, String>> dataSizeMsgs) {
        for (Map<String, String> yclRecord : docRecords) {
            Map<String, String> dataMap = new HashMap<>();
            dataMap.put("agency", yclRecord.get("agency"));
            dataMap.put("admin", yclRecord.get("admin"));
            dataMap.put("opinion_field", yclRecord.get("opinion_field"));
            dataMap.put("scope", yclRecord.get("scope"));
            dataMap.put("source_tunnel", yclRecord.get("source_tunnel"));
            dataMap.put("source_type", yclRecord.get("source_type"));
            dataMap.put("task_id", yclRecord.get("task_id"));
            dataMap.put("uid", yclRecord.get("uid"));
            dataMap.put("data_size", yclRecord.get("data_size"));
            dataMap.put("department", yclRecord.get("department"));
            dataSizeMsgs.add(dataMap);

            Map<String, String> recMap = new HashMap<>();
            recMap.put("admin", yclRecord.get("admin"));
            recMap.put("agency", yclRecord.get("agency"));
            recMap.put("opinion_field", yclRecord.get("opinion_field"));
            recMap.put("scope", yclRecord.get("scope"));
            recMap.put("source_tunnel", yclRecord.get("source_tunnel"));
            recMap.put("source_type", yclRecord.get("source_type"));
            recMap.put("task_id", yclRecord.get("task_id"));
            recMap.put("uid", yclRecord.get("uid"));
            recMap.put("rec_num", yclRecord.get("rec_num"));
            recMap.put("department", yclRecord.get("department"));
            recNumMsgs.add(recMap);
        }
    }
    
}
